package com.yunkazaixian.syntelecom.service.impl;

import cn.hutool.core.date.DateTime;
import cn.hutool.core.date.DateUtil;
import com.alibaba.fastjson.JSONObject;
import com.yunkazaixian.syntelecom.entity.HttpResponse;
import com.yunkazaixian.syntelecom.entity.McSimDx;
import com.yunkazaixian.syntelecom.entity.PushSession;
import com.yunkazaixian.syntelecom.entity.SessionInfo;
import com.yunkazaixian.syntelecom.mapper.McSimDxMapper;
import com.yunkazaixian.syntelecom.utils.R;
import com.yunkazaixian.syntelecom.utils.RedisUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Service;
import org.thymeleaf.util.StringUtils;

import java.time.ZoneId;
import java.time.format.DateTimeFormatter;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import java.util.stream.Collectors;

@Service
public class SessionService {
    private static final Logger log = LoggerFactory.getLogger(SessionService.class);
    private static final String SESSION_KEY_PREFIX = "sim_session:";
    private static final String PENDING_SESSIONS_KEY = "pending_sessions";
    private static final int BATCH_SIZE = 500;
    private static final long FLUSH_INTERVAL = 120000; // 30秒
    private static final long MAX_SESSION_DURATION = 86400000;
    private static final long DUPLICATE_WINDOW = 5000;
    private static final int MAX_LOCAL_CACHE_SIZE = 100000;
    private static final ZoneId UTC_ZONE = ZoneId.of("UTC");
    private static final DateTimeFormatter DATE_TIME_FORMATTER =
            DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss").withZone(UTC_ZONE);

    private static final String STATUS_ONLINE = "0";
    private static final String STATUS_OFFLINE = "1";

    @Autowired
    private RedisUtil redisUtil;
    @Autowired
    private McSimDxMapper mcSimDxMapper;
    @Autowired
    private ThreadPoolTaskExecutor asyncExecutor;

    private final ConcurrentHashMap<String, SessionInfo> localCache = new ConcurrentHashMap<>(MAX_LOCAL_CACHE_SIZE);
    private final AtomicInteger pendingCount = new AtomicInteger(0);
    private final AtomicLong lastFlushTime = new AtomicLong(System.currentTimeMillis());
    private final AtomicInteger activeSessions = new AtomicInteger(0);
    private final AtomicInteger totalRequests = new AtomicInteger(0);
    private final AtomicInteger duplicateRequests = new AtomicInteger(0);
    private final AtomicInteger filteredRequests = new AtomicInteger(0);

    // ======================== 开始会话处理 ========================
    public HttpResponse handleSessionStart(PushSession push) {
        totalRequests.incrementAndGet();
        if (push == null || StringUtils.isEmpty(push.getIccid())) {
            return R.fail("上送参数有误");
        }

        if (!"竣工".equals(push.getStatusInfo())) {
            return R.fail(push.getStatusInfo());
        }

        try {
            DateTime timestamp = DateUtil.parse(push.getData().getTimestamp());
            String iccid = push.getIccid();
            log.info("处理会话开始: ICCID={}, 时间={}", iccid, timestamp);

            String redisKey = SESSION_KEY_PREFIX + iccid;
            if (redisUtil.setIfAbsent(redisKey, "lock", DUPLICATE_WINDOW, TimeUnit.MILLISECONDS)) {
                updateLocalSession(iccid, timestamp, null, true);
                return R.ok();
            } else {
                duplicateRequests.incrementAndGet();
                return R.ok();
            }
        } catch (Exception e) {
            log.error("处理会话开始失败: {}", push.getIccid(), e);
            return R.fail("系统错误");
        }
    }

    // ======================== 结束会话处理 ========================
    public HttpResponse handleSessionEnd(PushSession push) {
        totalRequests.incrementAndGet();
        if (push == null || StringUtils.isEmpty(push.getIccid())) {
            return R.fail("上送参数有误");
        }

        if (!"竣工".equals(push.getStatusInfo())) {
            return R.fail(push.getStatusInfo());
        }

        String iccid = push.getIccid();
        DateTime timestamp = DateUtil.parse(push.getData().getTimestamp());

        // 查找会话（可能已被标记为 processed）
        SessionInfo session = localCache.values().stream()
                .filter(s -> s.getIccid().equals(iccid) && s.getStartTime() != null)
                .findFirst()
                .orElse(null);

        if (session == null) {
            log.warn("结束会话失败：未找到开始会话，ICCID={}", iccid);
            return R.fail("无效的会话结束请求：未找到开始会话");
        }

        try {
            String redisKey = SESSION_KEY_PREFIX + iccid;
            if (redisUtil.setIfAbsent(redisKey, "lock", DUPLICATE_WINDOW, TimeUnit.MILLISECONDS)) {
                // **关键修改：重置 processed 标记，确保加入队列**
                session.setProcessed(false);
                updateLocalSession(iccid, null, timestamp, false);
                return R.ok();
            } else {
                duplicateRequests.incrementAndGet();
                return R.ok();
            }
        } catch (Exception e) {
            log.error("处理会话结束失败: {}", iccid, e);
            return R.fail("系统错误");
        }
    }

    // ======================== 本地会话更新 ========================
    private void updateLocalSession(String iccid, DateTime startTime, DateTime endTime, boolean isStartEvent) {
        localCache.compute(iccid, (key, existing) -> {
            SessionInfo info = existing != null ? existing : new SessionInfo();
            info.setIccid(iccid);

            if (isStartEvent && startTime != null) {
                info.setStartTime(startTime);
                info.setEndTime(null);
                info.setStatus(STATUS_ONLINE);
                info.setProcessed(false); // 开始会话时标记为未处理
                log.debug("会话开始: ICCID={}, 开始时间={}", iccid, startTime);
            }

            if (!isStartEvent && endTime != null) {
                info.setEndTime(endTime);
                info.setStatus(STATUS_OFFLINE);
                // **关键修改：结束会话时重置 processed 标记**
                info.setProcessed(false);
                log.debug("会话结束: ICCID={}, 结束时间={}", iccid, endTime);
            }

            // **修改：无论是否处理过，只要有开始时间就加入队列**
            if (info.getStartTime() != null) {
                redisUtil.lPush(PENDING_SESSIONS_KEY, iccid);
                pendingCount.incrementAndGet();
                log.info("加入队列: ICCID={}, 状态={}, 处理标记={}",
                        iccid, info.getStatus(), info.isProcessed());
            }

            activeSessions.set(localCache.size());
            return info;
        });

        checkAndTriggerFlush();
    }

    // ======================== 批量处理触发逻辑 ========================
    private void checkAndTriggerFlush() {
        long now = System.currentTimeMillis();
        if (pendingCount.get() >= BATCH_SIZE || (now - lastFlushTime.get()) >= FLUSH_INTERVAL) {
            if (lastFlushTime.compareAndSet(lastFlushTime.get(), now)) {
                log.debug("触发批量处理: pendingCount={}, 距上次刷新={}ms",
                        pendingCount.get(), now - lastFlushTime.get());
                asyncExecutor.execute(this::flushPendingSessions);
            }
        }
    }

    // ======================== 批量处理执行 ========================
    @Scheduled(fixedRate = FLUSH_INTERVAL)
    public void flushPendingSessions() {
        long queueSize = redisUtil.lSize(PENDING_SESSIONS_KEY);
        log.info("批量处理检查: 队列大小={}, 本地缓存大小={}", queueSize, localCache.size());

        if (queueSize == 0) {
            log.debug("待处理队列为空，跳过批量处理");
            return;
        }

        log.info("开始批量处理待刷新会话, 队列大小: {}", queueSize);
        long startTime = System.currentTimeMillis();

        try {
            List<Object> iccidList = redisUtil.lRange(PENDING_SESSIONS_KEY, 0, BATCH_SIZE - 1);
            if (iccidList == null || iccidList.isEmpty()) {
                return;
            }

            processSessionsInBatches(iccidList);
            redisUtil.lTrim(PENDING_SESSIONS_KEY, iccidList.size(), -1);

            log.info("批量处理完成, 处理: {} 条记录, 耗时: {}ms",
                    iccidList.size(), System.currentTimeMillis() - startTime);
        } catch (Exception e) {
            log.error("批量处理会话异常", e);
        }
    }

    private void processSessionsInBatches(List<Object> iccidList) {
        for (int i = 0; i < iccidList.size(); i += 100) {
            List<String> iccids = iccidList.subList(i, Math.min(i + 100, iccidList.size()))
                    .stream()
                    .map(String::valueOf)
                    .collect(Collectors.toList());
            processSessionBatch(iccids);
        }
    }

    private void processSessionBatch(List<String> iccids) {
        List<McSimDx> records = new ArrayList<>(iccids.size());
        List<String> failedIccids = new ArrayList<>();

        for (String iccid : iccids) {
            try {
                SessionInfo session = localCache.get(iccid);
                if (session == null) {
                    log.warn("待处理会话不存在: {}", iccid);
                    continue;
                }

                // 标记为已处理，但不删除缓存
                session.setProcessed(true);
                pendingCount.decrementAndGet();

                if (session.getStartTime() == null) {
                    log.warn("无效会话（无开始时间）: {}", iccid);
                    continue;
                }

                // 过滤异常持续时间
                if (session.getEndTime() != null) {
                    long duration = session.getEndTime().getTime() - session.getStartTime().getTime();
                    if (duration <= 0 || duration >= MAX_SESSION_DURATION) {
                        filteredRequests.incrementAndGet();
                        log.warn("过滤异常会话（持续时间 {}ms）: {}", duration, iccid);
                        continue;
                    }
                }

                records.add(convertToEntity(session));
                log.debug("处理会话: ICCID={}, 状态={}", iccid, session.getStatus());
            } catch (Exception e) {
                log.error("处理会话异常: ICCID={}", iccid, e);
                failedIccids.add(iccid);
            }
        }

        if (!records.isEmpty()) {
            batchUpdateDatabase(records);
        }

        // 失败的 ICCID 重新入队
        if (!failedIccids.isEmpty()) {
            redisUtil.lPush(PENDING_SESSIONS_KEY, failedIccids.toArray());
            log.warn("重新加入失败的会话: {}", failedIccids);
        }
    }

    // ======================== 其他辅助方法 ========================
    private McSimDx convertToEntity(SessionInfo session) {
        McSimDx entity = new McSimDx();
        entity.setIccid(session.getIccid());
        entity.setCardboard(session.getStatus());

        if (session.getStartTime() != null) {
            entity.setImSaleProvince(session.getStartTime().toLocalDateTime().format(DATE_TIME_FORMATTER));
        }
        if (session.getEndTime() != null) {
            entity.setImSaleCity(session.getEndTime().toLocalDateTime().format(DATE_TIME_FORMATTER));
        }
        return entity;
    }

    private void batchUpdateDatabase(List<McSimDx> records) {
        try {
            mcSimDxMapper.batchUpdate(records);
        } catch (Exception e) {
            log.error("批量更新失败，尝试逐条处理", e);
            records.forEach(this::updateSingleRecord);
        }
    }

    private void updateSingleRecord(McSimDx record) {
        try {
            mcSimDxMapper.updateByIccid(record);
        } catch (Exception ex) {
            log.error("单条记录更新失败: {}", record.getIccid(), ex);
        }
    }

    // ======================== 状态监控 ========================
    public Map<String, Object> getSystemStatus() {
        Map<String, Object> status = new HashMap<>();
        status.put("activeSessions", activeSessions.get());
        status.put("pendingSessions", pendingCount.get());
        status.put("totalRequests", totalRequests.get());
        status.put("duplicateRequests", duplicateRequests.get());
        status.put("filteredRequests", filteredRequests.get());
        status.put("localCacheSize", localCache.size());
        status.put("redisQueueSize", redisUtil.lSize(PENDING_SESSIONS_KEY));
        return status;
    }
    // 增加队列监控
    @Scheduled(fixedRate = 60000)
    public void monitorQueue() {
        long queueSize = redisUtil.lSize(PENDING_SESSIONS_KEY);
        long cacheSize = localCache.size();

        // log.info("队列监控: 队列大小={}, 缓存大小={}", queueSize, cacheSize);
        if (queueSize > 10000) {
            log.warn("队列积压严重，当前大小: {}", queueSize);
            // 可添加告警逻辑，如发送邮件或短信
        }
    }
    // 新增定时清理缓存任务
    @Scheduled(fixedRate = 86400000)
    private void cleanExpiredSessions() {
        localCache.entrySet().stream()
                .filter(entry -> {
                    SessionInfo session = entry.getValue();
                    return session.getEndTime() != null &&
                            (System.currentTimeMillis() - session.getStartTime().getTime()) > MAX_SESSION_DURATION;
                })
                .limit(1000) // 每次最多清理1000条
                .forEach(entry -> localCache.remove(entry.getKey()));
    }
}